package com.linkstec.mongo;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.bson.Document;

import com.linkstec.util.ConfigUtil;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.client.FindIterable;
import com.mongodb.client.ListIndexesIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.IndexOptions;
import com.mongodb.client.model.Indexes;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.Updates;
import com.mongodb.client.result.UpdateResult;

public class MongoDbUtil {

	private static final Logger logger = LogManager.getLogger();
	private static String hostsAndPorts;
	private static String userNameDatabasePassword;
	private static MongoClient mongoClient = null;
	private static MongoDatabase testDatabase;
	private static MongoCollection<Document> collection;

	/*
	 * @SuppressWarnings("rawtypes") public static void setConfig(Map cfg) {
	 * hostsAndPorts = (String) cfg.get("mongoHostsAndPorts"); }
	 */
	static {
		hostsAndPorts = "192.168.2.232:30000,192.168.2.233:30000,192.168.2.234:30000";
		userNameDatabasePassword = ConfigUtil.getProperty("userNameDatabasePassword", "mot,MOT,Gtjamot@123");
	}

	// connect to mongodb
	@SuppressWarnings("deprecation")
	public static MongoClient getMongoClient() {
		if (null == mongoClient) {

			MongoClientOptions.Builder build = new MongoClientOptions.Builder();
			build.connectionsPerHost(50); // 与目标数据库能够建立的最大connection数量为50
			// 如果当前所有的connection都在使用中，则每个connection上可以有50个线程排队等待
			build.threadsAllowedToBlockForConnectionMultiplier(50);
			/*
			 * 一个线程访问数据库的时候，在成功获取到一个可用数据库连接之前的最长等待时间为2分钟
			 * 这里比较危险，如果超过maxWaitTime都没有获取到这个连接的话，该线程就会抛出Exception
			 * 故这里设置的maxWaitTime应该足够大，以免由于排队线程过多造成的数据库访问失败
			 */
			build.maxWaitTime(1000 * 60 * 2);
			build.connectTimeout(1000 * 60 * 1); // 与数据库建立连接的timeout设置为1分钟
//			build.socketKeepAlive(true);//最关键，保证长时间不用不出错

			MongoClientOptions myOptions = build.build();

			// mongodb集群
			List<ServerAddress> addresses = new ArrayList<ServerAddress>();
			for (String hostAndPort : hostsAndPorts.split(",")) {
				String host = hostAndPort.split(":")[0];
				int port = Integer.valueOf(hostAndPort.split(":")[1]);
				ServerAddress address = new ServerAddress(host, port);
				addresses.add(address);
			}
			if (StringUtils.isBlank(userNameDatabasePassword)) {
				logger.info("userNameDatabasePassword为空白");
				mongoClient = new MongoClient(addresses, myOptions);
			} else {
				logger.info("userNameDatabasePassword为" + userNameDatabasePassword);
				String[] Credential = userNameDatabasePassword.split(",");
				MongoCredential credential = MongoCredential.createCredential(Credential[0], Credential[1],
						Credential[2].toCharArray());
				mongoClient = new MongoClient(addresses, credential, myOptions);
			}

			return mongoClient;
		} else {
			return mongoClient;
		}
	}

	public static MongoDatabase getMongoDatabase(String mongoDatabase) {
		mongoClient = getMongoClient();
		testDatabase = mongoClient.getDatabase(mongoDatabase);
		logger.info("获得testDatabase");
		return testDatabase;
	}

	public static MongoCollection<Document> getMongoCollection(MongoDatabase mongoDatabase, String collectionName) {
		collection = mongoDatabase.getCollection(collectionName);
		return collection;
	}

	public static MongoCollection<Document> getMongoCollection(String mongoDatabase, String collectionName) {
		mongoClient = getMongoClient();
		testDatabase = mongoClient.getDatabase(mongoDatabase);
		collection = testDatabase.getCollection(collectionName);
		return collection;
	}
	public static void createIndex() {
		MongoCollection<Document> motEventRepeat = getMongoCollection("MOT", "MOT_EVENT_REPEAT");
		ListIndexesIterable<Document> listIndexesIterable = motEventRepeat.listIndexes();
		String indexName = "createTime";
		Long expireAfter = 100L;
		boolean indexExists = false;
		MongoCursor<Document> mongoCursorIndexes = listIndexesIterable.iterator();
		while (mongoCursorIndexes.hasNext()) {
			Document document = mongoCursorIndexes.next();
			if (document.containsValue(indexName + "_1")
					&& expireAfter.longValue() == document.getInteger("expireAfterSeconds").longValue()) {
				logger.info(indexName + "索引已存在;过期时间" + expireAfter + ",无需创建");
				indexExists = true;
			} else if (document.containsValue(indexName + "_1")
					&& expireAfter.longValue() != document.getInteger("expireAfterSeconds").longValue()) {
				indexExists = true;
				motEventRepeat.dropIndex(indexName + "_1");
				// expireAfter秒自动删除
				IndexOptions indexOptions = new IndexOptions().expireAfter(expireAfter, TimeUnit.SECONDS);
				motEventRepeat.createIndex(Indexes.ascending(indexName), indexOptions);
				logger.info(indexName + "索引重建成功;原过期时间" + document.getInteger("expireAfterSeconds").longValue() + "新过期时间"
						+ expireAfter);
			}
		}
		if (!indexExists) {
			// expireAfter秒自动删除
			IndexOptions indexOptions = new IndexOptions().expireAfter(expireAfter, TimeUnit.SECONDS);
			motEventRepeat.createIndex(Indexes.ascending(indexName), indexOptions);
			logger.info(indexName + "索引创建成功;过期时间" + expireAfter);
		}
	}
	public static void insertOne() {
		MongoCollection<Document> users = getMongoCollection("MOT", "users");
		Document user = new Document();
		user.put("name", "jimmy");
		user.put("age", "34");
		user.put("工作", "程序员");
		Document address = new Document();
		address.put("city", "bj");
		address.put("street", "bq road");
		address.put("mail", "ufpark 68#");
		/**
		 * 对于内嵌文档，我们需要先将内嵌文档填充后，再填充到外层文档中
		 */
		user.put("address", address);
        users.insertOne(user);

		logger.info("" + users.countDocuments());
	}
	public static void updateOne() {
		MongoCollection<Document> motEventRepeat = getMongoCollection("MOT", "MOT_EVENT_REPEAT");
		UpdateOptions updateOptions = new UpdateOptions();
		updateOptions.upsert(true);
		UpdateResult updateResult = motEventRepeat
				.updateOne(Filters.and(Filters.eq("event_id", "382"), Filters.eq("customer_id", "12345678")),
						Updates.combine(Updates.setOnInsert("event_id", "382"),
								Updates.setOnInsert("customer_id", "12345678"),
								Updates.setOnInsert("createTime", new Date())),
						updateOptions);
		logger.info(updateResult.getMatchedCount() + "");
		// 取出查看
		FindIterable<Document> iterable = motEventRepeat.find();
		MongoCursor<Document> mongoCursor = iterable.iterator();
		while (mongoCursor.hasNext()) {
			Document EventRepeat = mongoCursor.next();
			logger.info(EventRepeat.toJson());
			// 日期类型自动转化了
			logger.info(EventRepeat.getDate("createTime").toString());
		}
	}

	public static void main(String[] args) {

		createIndex();
		insertOne();
		logger.info("----------------------------------华丽的分割线-------------------------------------------------");
		updateOne();
		logger.info("----------------------------------华丽的分割线-------------------------------------------------");
		System.exit(0);
		long start = System.currentTimeMillis();
		int availableProcessors = Runtime.getRuntime().availableProcessors();
		int poolSize = availableProcessors / 2;
		ExecutorService es = Executors.newFixedThreadPool(poolSize);
		logger.info("初始化线程池大小：" + poolSize);
		List<Future<?>> list = new ArrayList<Future<?>>();
		MongoCollection<Document> smotUserMessageConfig = MongoDbUtil.getMongoCollection("MOT",
				"SMOT_USER_MESSAGE_CONFIG");
		for (int i = 0; i < poolSize; i++) {
			list.add(es.submit(new UpdateOneCallable(smotUserMessageConfig)));
		}
		
		int size = 2000000;
		UpdateOptions updateOptions = new UpdateOptions();
		updateOptions.upsert(true);
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
		for (int i = 0; i < size; i++) {
			UpdateOneModel<Document> updateOneModel = new UpdateOneModel<Document>(
					Filters.and(Filters.eq("userId", "018637"), Filters.eq("eventId", "455" + i),
							Filters.eq("toDay", sdf.format(new Date()))),
					Updates.combine(Updates.setOnInsert("userId", "018637"),
							Updates.setOnInsert("eventId", "455" + i), Updates.set("isNotice", "1"),
							Updates.set("isSms", "1"), Updates.set("isEmail", "1"), Updates.set("isApp", "1"),
							Updates.set("isWebmessage", "1"), Updates.set("isWechat", "1"),
							Updates.setOnInsert("toDay", sdf.format(new Date())),
							Updates.setOnInsert("createTime", new Date())),
					updateOptions);
			UpdateOneCallable.put(updateOneModel);
		}
		
		Document document = new Document();
		for (int i = 0; i < list.size(); i++) {
			UpdateOneCallable.put(new UpdateOneModel<>(
					document,
					document));
		}
		for (int i = 0; i < list.size(); i++) {
			try {
				list.get(i).get();
			} catch (InterruptedException e) {
				logger.error("", e);
			} catch (ExecutionException e) {
				logger.error("", e);
			}
		}
		es.shutdown();
		long end = System.currentTimeMillis();
		logger.info((end - start) + "");

	}
}

class UpdateOneCallable implements Callable<Object> {
	
	private static final Logger logger = LogManager.getLogger();
	private static BlockingQueue<UpdateOneModel<Document>> blockingQueue = new LinkedBlockingDeque<UpdateOneModel<Document>>(
			1000);
	int i = 0;
	private MongoCollection<Document> collection;
	public static void put(UpdateOneModel<Document> updateOneModel) {
		try {
			blockingQueue.put(updateOneModel);
		} catch (InterruptedException e) {
			logger.error("", e);
		}
	}

	public UpdateOneCallable(MongoCollection<Document> collection) {
		super();
		this.collection = collection;
	}

	@Override
	public Object call() {
		List<UpdateOneModel<Document>> updateOneModelList = new ArrayList<UpdateOneModel<Document>>();
		UpdateOneModel<Document> updateOneModel = null;
		while (true) {
			try {
				updateOneModel = blockingQueue.take();
				
				if (updateOneModel.getUpdate() instanceof Document) {
					if (updateOneModelList.size() > 0) {
						collection.bulkWrite(updateOneModelList);
						updateOneModelList.clear();
						updateOneModelList = new ArrayList<UpdateOneModel<Document>>();
					}
					logger.info("共更新数据量：" + i);
					break;
				}
				i++;
				updateOneModelList.add(updateOneModel);
				if (updateOneModelList.size() >= 1000) {
					collection.bulkWrite(updateOneModelList);
					updateOneModelList.clear();
					updateOneModelList = new ArrayList<UpdateOneModel<Document>>();
				}
			} catch (InterruptedException e) {
				logger.error("", e);
			}
		}
		return null;
	}
}